package com.dahua.messagesyncsh.kafka.runner;

import com.alibaba.fastjson2.JSONObject;
import com.dahua.messagesyncsh.config.redis.RedisUtil;
import com.dahua.messagesyncsh.criteria.CollectionQueryCriteria;
import com.dahua.messagesyncsh.entity.Collection;
import com.dahua.messagesyncsh.entity.SensorData;
import com.dahua.messagesyncsh.pojo.CMode;
import com.dahua.messagesyncsh.pojo.EquipmentInfo;
import com.dahua.messagesyncsh.service.CollectionService;
import com.dahua.messagesyncsh.service.EquipmentService;
import com.dahua.messagesyncsh.service.Impl.SensorDataServiceImpl;
import com.dahua.messagesyncsh.service.ProcessorService;
import com.dahua.messagesyncsh.service.SensorDataService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;

@Slf4j
@Component
public class CModeHandleRunner implements Runnable {
    @Resource
    private RedisUtil redisUtil;

    @Resource
    private RedissonClient redissonClient;

    @Resource
    private EquipmentService equipmentService;

    @Resource
    private SensorDataService sensorDataService;

    @Resource
    private ProcessorService processorService;

    @Resource
    private CollectionService collectionService;

    private final List<ConsumerRecord<String, String>> consumerRecords;

    public CModeHandleRunner(RedisUtil redisUtil,
                             RedissonClient redissonClient,
                             EquipmentService equipmentService,
                             SensorDataService sensorDataService,
                             ProcessorService processorService,
                             CollectionService collectionService,
                             List<ConsumerRecord<String, String>> consumerRecords) {
        this.redisUtil = redisUtil;
        this.redissonClient = redissonClient;
        this.equipmentService = equipmentService;
        this.sensorDataService = sensorDataService;
        this.processorService = processorService;
        this.consumerRecords = consumerRecords;
        this.collectionService = collectionService;
    }

    @Override
    public void run() {
        this.handle(consumerRecords);
    }

    public void handle(List<ConsumerRecord<String, String>> consumerRecords) {
        for (ConsumerRecord<String, String> item : consumerRecords) {
            CMode cMode = null;
            try {
                cMode = JSONObject.parseObject(item.value(), CMode.class);
            } catch (Exception e) {
                log.error("【CMode】：item转换错误 \n {}", e.toString());
            }
            if (cMode == null) {
                return;
            }

            if (cMode.getEquipmentId() == null || cMode.getEquipmentId().isEmpty() ||
                    cMode.getSensorId() == null || cMode.getSensorId().isEmpty() ||
                    cMode.getStartTime() == null ||
                    cMode.getType() == null || cMode.getType().isEmpty() ||
                    cMode.getHeader() == null || cMode.getHeader().length == 0 ||
                    cMode.getData() == null){
                log.error("【CMode】：item数据不完整 \n {}", cMode);
                continue;
            }

            /*  设备不存在*/
            EquipmentInfo enquire = equipmentService.enquire(cMode.getEquipmentId());

            if (enquire == null) {
                continue;
            }

            /*  通过对方测点获取我方测点id*/
            Collection collection = null;
            List<Collection> collections = collectionService.enquire(CollectionQueryCriteria.builder().collection(enquire.getBk82()).sensor(cMode.getSensorId()).build());
            if (collections.size() != 1) {
                continue;
            }else{
                collection = collections.get(0);
            }

            if (collection == null || collection.getSensorId() == null){
                log.error("【CMode】：测点对应不存在 \n {}", cMode.getSensorId());
                continue;
            }

            /*  一个设备的一个时间段的数据争用*/
            String key = cMode.getEquipmentId() + "." + collection.getSensorId() + "." + cMode.getStartTime();
            String lockKey = "lock_" + key;
            RLock lock = redissonClient.getLock(lockKey);
            try {
                lock.lock();

                SensorData sensorData = processorService.buildData(cMode, enquire, collection);
                if (sensorData == null) {
                    continue;
                }
                try{
                    sensorDataService.toInsert(key,cMode,sensorData, SensorDataServiceImpl.Type.CHART_FILE);
                }catch (Exception e){
                    log.error("【CMode】：数据更新处理异常 \n{}", e.getMessage());
                }

            } catch (Exception e) {
                log.error("【CMode】：第一次锁处理异常\n {}", e.toString());
            } finally {
                lock.unlock();
            }
        }
    }
}
